/*
 *  Queen Mary University of London MOM System
 *  
 *  Copyright Queen Mary University of London
 *  
 *  Authors:
 *  Bob Chew (bob.chew@elec.qmul.ac.uk)
 *  Beatriz Viñal Murciano (beavimu@gmail.com)
 *  John Bigham (john.bigham@elec.qmul.ac.uk)
 *  
 *  
 *  Portions of this software uses Apache Qpid
 *    http://qpid.apache.org
 *  
 *  Apache Qpid license:
 * --------------------------------------------------------------
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 * -------------------------------------------------------------
 * 
 */
package overlayManager;

import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Properties;
import java.util.Set;

import javax.jms.JMSException;
import javax.jms.TopicSubscriber;

import momControlClient.MOMControlClient;

import org.apache.qpid.AMQException;
import org.apache.qpid.client.TopicPublisherAdapter;
import org.apache.qpid.url.URLSyntaxException;


/**
 * Create all the elements in the broker failure detection system.
 * Assign a broker to new clients that are created.
 * Reassign a broker to clients who are using a broker that fails.
 */
public class OverlayManager extends MOMControlClient {

    /** Topic used to receive messages from clients. */
    private static final String LISTEN_TOPIC = "brokerRequests";
    /**
     * Root of the topic used to send update messages to clients (the client ID
     * must be added).
     */
    private static final String UPDATES_TOPIC_ROOT = "brokerUpdates.";
    /**
     * Root of the topic used to send assignment messages to clients (the
     * client ID must be added).
     */
    private static final String ASSIG_TOPIC_ROOT = "brokerAssig.";
    /** Topic used to receive messages from controller. */
    private static final String TOPIC_OM = "brokerFailure.overlayManager";

    /** IP address of the controller. */
    private String controllerIP;
    /** Number of brokers in the system. */
    private int numBrokers;
    /**
     * List of IP addresses of the machines in which brokers are running and
     * the ports in which they are listening.
     */
    private HashMap<String, Integer> brokers = new HashMap<String, Integer>();

    /** Properties of the topology of the network. */
    private Properties topologyProps;
    /** Name of the file in which the properties are saved. */
    private static final String TOP_FILE_NAME = "nwTopology.xml";
//    //description of the file
//    private final String TOP_FILE_DESC = "This file contains the data that "
//            + "allows elements in the system to know where to find other "
//            + "elements in the system.";

    /** Properties of brokers assignment. */
    private Properties pubAssignments = new Properties();
    /** Name of the file in which the properties are saved. */
    private static final String PUB_ASSIG_FILE_NAME = "brokerPublishers.xml";
    /** Description of the file. */
    private static final String PUB_ASSIG_FILE_DESC = "This file contains data"
            + " about brokers assigned to publishers.";

    /** Properties of subscribers assignment. */
    private Properties subAssignments = new Properties();
    /** Name of the file in which the properties are saved. */
    private static final String SUB_ASSIG_FILE_NAME = "brokerSubscribers.xml";
    /** Description of the file. */
    private static final String SUB_ASSIG_FILE_DESC = "This file contains data"
            + " about brokers assigned to subscribers.";

    /**
     * List of machines in which a broker is running and their state.
     * State is an int from WORST_STATE (worst) to BEST_STATE (best)
     */
    private static HashMap<String, Integer> brokersState
            = new HashMap<String, Integer>();
    /** Best state of a broker. */
    private static final int BEST_STATE = 10;
    /** Worst state of a broker. */
    private static final int WORST_STATE = 0;
    /** Threshold state of a broker (above it, the broker is working well). */
    private static final int TH_STATE = (WORST_STATE + BEST_STATE) / 2;


    /**
     * Constructor.
     * Read information about the network from PROPS_FILE_NAME, create the
     * elements in the broker failure detection system and start listening
     * to messages from the controller or new clients.
     * @throws JMSException  if there is any problem
     * @throws AMQException  if there is any problem
     * @throws URLSyntaxException  if there is any problem
     */
    public OverlayManager()
            throws JMSException, URLSyntaxException, AMQException {

        super(); //assign ID

        //read properties from properties file
        try {
            FileInputStream fist = new FileInputStream(TOP_FILE_NAME);
            topologyProps = new Properties();
            topologyProps.loadFromXML(fist);
            controllerIP = topologyProps.getProperty("controllerIP");
            numBrokers = Integer.parseInt(
                    topologyProps.getProperty("numBrokers"));
            for (int i = 1; i <= numBrokers; i++) {
                String ip = "broker" + i + "IP";
                String port = "broker" + i + "port";
                String brokerIP = topologyProps.getProperty(ip);
                int brokerPort
                        = Integer.parseInt(topologyProps.getProperty(port));
                brokers.put(brokerIP, brokerPort);
                //set broker state to WORST_STATE
                brokersState.put(brokerIP, WORST_STATE);
            }
            System.out.println("Controller IP: " + controllerIP);
            System.out.println("Number of brokers: " + numBrokers);
            System.out.println("Brokers IPs: " + brokers);
        } catch (Exception e) {
            e.printStackTrace();
        }

        //create and start a connection to the controller broker and
        //a session in it
        openConnAndSess(controllerIP);

        //create a subscriber for the topic
        TopicSubscriber clSub = createSubscriber(LISTEN_TOPIC);
        //set its message listener
        clSub.setMessageListener(new RequestMessageListener(this));

        System.out.println("Overlay manager waiting for messages in "
                           + LISTEN_TOPIC);

        //create a subscriber for the topic
        TopicSubscriber ctrlSub = createSubscriber(TOPIC_OM);
        //set its message listener
        ctrlSub.setMessageListener(new ControllerMessageListener(this));

        System.out.println("Overlay manager waiting for messages in "
                           + TOPIC_OM);

    }

    /**
     * Set the state of a broker.
     *
     * @param broker  whose state will be set
     * @param state  new state of the broker
     */
    protected void setBrokerState(final String broker, final int state) {
        brokersState.put(broker, state);
        System.out.println("State in the overlay manager: " + brokersState);
    }

    /**
     * Assign primary and mirror broker for the topics of a publisher and send
     * the assignment to the publisher.
     *
     * The format of the assignment is:
     * topic1:primaryBroker1:primaryPort1:mirrorBroker1:mirrorPort1:topic2:...
     *
     * @param pId  client ID of the publisher
     * @param topics  topics of the publisher
     * @throws JMSException  if there is any problem
     */
    protected void assignBrokersToPub(final String pId, final String[] topics)
            throws JMSException {

        //get the assignment
        String assignment = getBrokersForPub(pId, topics);

        //create name of the topic to which the publisher is listening
        String topicName = ASSIG_TOPIC_ROOT + pId;
        //create publisher for topic
        TopicPublisherAdapter publisher = createPublisher(topicName);
        //send the message to the publisher
        sendMessage(publisher, assignment);

        System.out.println("Overlay manager assignment: " + assignment
                           + " to publisher " + pId);

    }

    /**
     * Get primary and mirror broker for the topics of a publisher.
     *
     * The format of the assignment is:
     * topic1:primaryBroker1:primaryPort1:mirrorBroker1:mirrorPort1:topic2:...
     *
     * @param pId  client ID of the publisher
     * @param topics  topics of the publisher
     * @return  string with the assignments to send to the publisher
     */
    private String getBrokersForPub(final String pId, final String[] topics) {

        //choose two working brokers
        //TODO: assign different brokers to different topics
        //so far we assign the same brokers to all the topics of a publisher
        String primaryBroker = getWorkingBroker();
        String mirrorBroker = getAnotherWorkingBroker(primaryBroker);

        String assignment = "";

        for (String topic : topics) {
            //add the assigned brokers for this topic to the assignment
            assignment = assignment + topic + ":" + primaryBroker + ":"
                         + brokers.get(primaryBroker) + ":" + mirrorBroker + ":"
                         + brokers.get(mirrorBroker) + ":";

            //save the assignments (no entry if they are null)
            String primaryProp = pId + ":" + topic + ":primary";
            String mirrorProp = pId + ":" + topic + ":mirror";
            if (primaryBroker != null) {
                pubAssignments.setProperty(primaryProp, primaryBroker);
            }
            if (mirrorBroker != null) {
                pubAssignments.setProperty(mirrorProp, mirrorBroker);
            }
        }

        //save the properties file
        try {
            FileOutputStream fos = new FileOutputStream(PUB_ASSIG_FILE_NAME);
            pubAssignments.storeToXML(fos, PUB_ASSIG_FILE_DESC);
        } catch (Exception e) {
            e.printStackTrace();
        }

        //return the assignment
        return assignment;

    }

    /**
     * Get brokers for the topics of a subscriber and send the assignment to
     * the subscriber.
     *
     * The format of the assignment is:
     * topic1:primaryBroker1:primaryPort1:mirrorBroker1:mirrorPort1:topic2...
     *
     * @param sId  client ID of the subscriber
     * @param topics  topics of the subscriber
     * @throws JMSException  if there is any problem
     */
    protected void assignBrokersToSub(final String sId, final String[] topics)
            throws JMSException {

        //get the assignment
        String assigment = getBrokersForSub(sId, topics);

        //create name of the topic to which the subscriber is listening
        String topicName = ASSIG_TOPIC_ROOT + sId;
        //create publisher for topic
        TopicPublisherAdapter publisher = createPublisher(topicName);
        //send the message to the subscriber
        sendMessage(publisher, assigment);

        System.out.println("Overlay manager assignment: " + assigment
                           + " to subscriber " + sId);

    }

    /**
     * Get brokers for the topics of a subscriber.
     *
     * The format of the assignment is:
     * topic1:primaryBroker1:primaryPort1:mirrorBroker1:mirrorPort1:topic2...
     *
     * @param sId  client ID of the subscriber
     * @param topics  topics of the subscriber
     * @return  string with the assignments to send to the subscriber
     */
    private String getBrokersForSub(final String sId, final String[] topics) {

        String assignment = "";

        //look for the brokers where topics are being published to
        for (String topic : topics) {
            //get the primary and mirror broker of the topic
            String primaryBroker = getPrimaryBroker(topic);
            String mirrorBroker = getMirrorBroker(topic);

            //add the brokers for the topic to the assignment
            assignment = assignment + topic + ":" + primaryBroker + ":"
                         + brokers.get(primaryBroker) + ":" + mirrorBroker
                         + ":" + brokers.get(mirrorBroker) + ":";
            //save the assignments (no entry if they are null)
            String primaryProp = sId + ":" + topic + ":primary";
            String mirrorProp = sId + ":" + topic + ":mirror";
            if (primaryBroker != null) {
                subAssignments.setProperty(primaryProp, primaryBroker);
            }
            if (mirrorBroker != null) {
                subAssignments.setProperty(mirrorProp, mirrorBroker);
            }
        }

        //save the properties file
        try {
            FileOutputStream fos = new FileOutputStream(SUB_ASSIG_FILE_NAME);
            subAssignments.storeToXML(fos, SUB_ASSIG_FILE_DESC);
        } catch (Exception e) {
            e.printStackTrace();
        }

        //return the assignment
        return assignment;

    }

    /**
     * Switch the broker for a topic.
     *
     * @param pubTopic  topic to move
     * @param pubId  ID of the publisher of the topic
     * @param pubType  type of broker the publisher was using it as
     *                    (primary or mirror)
     *
     * @throws IOException  if there is any problem
     * @throws JMSException  if there is any problem
     */
    protected void switchBroker(final String pubTopic, final String pubId,
            final String pubType) throws IOException, JMSException {

        String primaryPubProp = pubId + ":" + pubTopic + ":primary";
        String mirrorPubProp = pubId + ":" + pubTopic + ":mirror";

        //assign another broker to the publisher for that topic
        String otherPubBroker; //other broker in use
        String newPubBroker; //new assigned broker

        if (pubType.equals("mirror")) {
            otherPubBroker = pubAssignments.getProperty(primaryPubProp);
            newPubBroker = getAnotherWorkingBroker(otherPubBroker);
            //store the assignment (it is always a mirror broker)
            if (newPubBroker != null) {
                pubAssignments.setProperty(mirrorPubProp, newPubBroker);
            }
        } else { //type.equals("primary")
            otherPubBroker = pubAssignments.getProperty(mirrorPubProp);
            newPubBroker = getAnotherWorkingBroker(otherPubBroker);
            //save the assignment (now old mirror is primary)
            if (otherPubBroker != null) {
                pubAssignments.setProperty(primaryPubProp, otherPubBroker); 
            }
            //the assignment is always a mirror broker
            if (newPubBroker != null) {
                pubAssignments.setProperty(mirrorPubProp, newPubBroker);
            }
        }

        //save the properties file
        FileOutputStream fosP = new FileOutputStream(PUB_ASSIG_FILE_NAME);
        pubAssignments.storeToXML(fosP, PUB_ASSIG_FILE_DESC);

        //name of the topic to which the publisher is listening
        String pubTopicName = UPDATES_TOPIC_ROOT + pubId;
        //create publisher for topic
        TopicPublisherAdapter pubPublisher = createPublisher(pubTopicName);
        //create text of the message
        String pubAssignment = pubTopic + ":" + pubType + ":" + newPubBroker
                               + ":" + brokers.get(newPubBroker);
        //send the message to the publisher
        sendMessage(pubPublisher, pubAssignment);

        System.out.println("Overlay manager reassignment: " + pubAssignment
                           + " to publisher " + pubId);

        //find the subscribers subscribed to it
        Set<String> subsToTopic = getSubsToTopic(pubTopic);
        for (String sId : subsToTopic) {
            //get topic new brokers
            String newSubPrimaryBroker = getPrimaryBroker(pubTopic);
            String newSubMirrorBroker = getMirrorBroker(pubTopic);

            //save the assignment
            String primarySubProp = sId + ":" + pubTopic + ":primary";
            String mirrorSubProp = sId + ":" + pubTopic + ":mirror";
            if (newSubPrimaryBroker != null) {
                subAssignments.setProperty(primarySubProp, newSubPrimaryBroker);
            }
            if (newSubMirrorBroker != null) {
                subAssignments.setProperty(mirrorSubProp, newSubMirrorBroker);
            }
            //save the properties file
            FileOutputStream fosS = new FileOutputStream(SUB_ASSIG_FILE_NAME);
            subAssignments.storeToXML(fosS, SUB_ASSIG_FILE_DESC);

            //name of the topic to which the subscriber is listening
            String topicSubName = UPDATES_TOPIC_ROOT + sId;
            //create publisher for topic
            TopicPublisherAdapter publisherSub = createPublisher(topicSubName);
            //create text of the message
            String subAssignment = pubTopic + ":" + pubType + ":"
                                   + newSubMirrorBroker + ":"
                                   + brokers.get(newSubMirrorBroker);
            //send the message to the subscriber
            sendMessage(publisherSub, subAssignment);

            System.out.println("Overlay manager reassignment: "
                               + subAssignment + " to subscriber " + sId);
        }

    }

    /**
     * Get a broker that is working.
     *
     * @return  IP address of a broker that is working
     */
    private String getWorkingBroker() {
        Set<String> brks = brokersState.keySet();
        for (String broker : brks) { //look for a working broker
            if (brokersState.get(broker) > TH_STATE) {
                //one working broker found
                return broker;
            }
        }
        return null;
    }

    /**
     * Get a broker that is working different than the given one.
     *
     * @param brkr  broker not to choose
     * @return  IP address of a broker that is working and is not the given one
     */
    private String getAnotherWorkingBroker(final String brkr) {
        Set<String> brks = brokersState.keySet();
        for (String broker : brks) { //look for a working broker
            if (brokersState.get(broker) > TH_STATE && !broker.equals(brkr)) {
                //a different working broker found
                return broker;
            }
        }
        return null;
    }

    /**
     * Get the primary broker of a topic.
     *
     * @param topicName  topic whose primary broker we want to find
     * @return  IP address of the primary broker of the topic or null if there is
     *             no primary broker assigned to it
     */
    private String getPrimaryBroker(final String topicName) {
        Set<String> propNames = pubAssignments.stringPropertyNames();
        for (String propName : propNames) {
            //if it is the primary broker
            if (propName.endsWith(topicName + ":primary")) {
                return pubAssignments.getProperty(propName);
            }
        }
        return null;
    }

    /**
     * Get the mirror broker of a topic.
     *
     * @param topicName  topic whose mirror broker we want to find
     * @return  IP address of the mirror broker of the topic or null if there
     *             is no mirror broker assigned to it
     */
    private String getMirrorBroker(final String topicName) {
        Set<String> propNames = pubAssignments.stringPropertyNames();
        for (String propName : propNames) {
            //if it is the mirror broker
            if (propName.endsWith(topicName + ":mirror")) {
                return pubAssignments.getProperty(propName);
            }
        }
        return null;
    }

    /**
     * Get the property names of the topics that are being published to the
     * given broker.
     *
     * @param broker  to look for topics published to it
     * @return  set of prop names of topics published to the given broker
     */
    protected Set<String> getTopicsPubToBroker(final String broker) {
        Set<String> topics = new HashSet<String>();

        Set<String> propPubNames = pubAssignments.stringPropertyNames();

        for (String propPubName : propPubNames) {
            //if a publisher was using that broker
            if (((String) pubAssignments.get(propPubName)).equals(broker)) {
                topics.add(propPubName);
            }
        }
        return topics;
    }

    /**
     * Get the subscribers to the given topic.
     *
     * @param topic  to look for subscribers to it
     * @return  set of IDs of subscribers to the given topic
     */
    private Set<String> getSubsToTopic(final String topic) {
        Set<String> subProps = new HashSet<String>();

        Set<String> propSubNames = subAssignments.stringPropertyNames();

        for (String propSubName : propSubNames) {
            String[] parts = propSubName.split(":");
            //if a subscriber was subscribed to that topic
            if (parts[1].equals(topic)) {
                subProps.add(parts[0]);
            }
        }
        return subProps;
    }

    /**
     * Get the publisher to the given topic in the given broker.
     *
     * @param topic  to look for its publisher
     * @param broker  in which the topic is being published to
     * @return  prop name of the publisher to the given topic and broker or
     *          null if it can not find it
     */
    protected String getPubToTopicAndBroker(final String topic,
                                            final String broker) {

        Set<String> propPubNames = pubAssignments.stringPropertyNames();

        for (String propPubName : propPubNames) {
            //if a publisher was using that broker for that topic
            if (propPubName.contains(topic)
                && ((String) pubAssignments.get(propPubName)).equals(broker)) {
                return propPubName;
            }
        }

        return null;

    }

    /**
     * Create a new overlay manager.
     *
     * @param args  array of arguments from the command line (unused)
     * @throws JMSException  if there is any problem
     * @throws AMQException  if there is any problem
     * @throws URLSyntaxException  if there is any problem
     */
    public static void main(final String[] args)
            throws JMSException, URLSyntaxException, AMQException {
        new OverlayManager();
    }

}
